package com.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.dto.CreateSeckillOrderRequest;
import com.mall.promo.cache.CacheManager;
import com.mall.promo.dal.persistence.PromoItemMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;

/**
 * @Description: 发送事务消息
 * @Author: 徐能雄
 * @Date: 2021/10/7 20:25
 */
@Component
@Slf4j
public class CreatePromoOrderProducer {

    private TransactionMQProducer transactionMQProducer;

    @Autowired
    PromoItemMapper promoItemMapper;

    @Autowired
    CacheManager cacheManager;

    @Autowired
    RedissonClient redissonClient;

    @PostConstruct
    public void init(){

        //输出日志
        //重要！重要！重要！
        //便于排查bug
        log.info("CreatePromoOrderProducer init start ...");

        //1.创建一个用于发送事务消息的producer对象
        transactionMQProducer = new TransactionMQProducer("create_promo_order_producer");

        //2.设置nameserver地址
        transactionMQProducer.setNamesrvAddr("127.0.0.1:9876");

        //3.设置事务监听器
        transactionMQProducer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object arg) {

                Map<String, Long> parmMap = (Map<String, Long>) arg;
                Long productId = parmMap.get("productId");
                Long psId = parmMap.get("psId");


                Integer effectiveRow = 0;
                //从redis中获取一把锁
                String lockKey = "create_promo_stock_" + productId + "_" + psId;
                RLock lock = redissonClient.getLock(lockKey);

                //分布式锁同时只能被一个线程加锁成功，加锁失败的线程会阻塞等待
                lock.lock();
                try {
                    //扣减库存
                    effectiveRow = promoItemMapper.decreaseStock(productId, psId);

                }finally {
                    lock.unlock();
                }

                String localTransactionKey = "local_transaction_" + message.getTransactionId();
                if (effectiveRow < 1) {
                    //没有库存了
                    //将结果保存至redis，这里设置过期时间（expire）为1天
                    cacheManager.setCache(localTransactionKey, "rollback", 1);
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                //还有库存
                //将结果保存至redis，这里设置过期时间（expire）为1天
                cacheManager.setCache(localTransactionKey, "commit", 1);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {

                String localTransactionKey = "local_transaction_" + messageExt.getTransactionId();
                String transactionResult = cacheManager.checkCache(localTransactionKey);

                if (transactionResult != null && "commit".equals(transactionResult.trim())) {
                    //说明有库存，本地事务执行成功
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                if (transactionResult != null && "rollback".equals(transactionResult.trim())) {
                    //没有有库存，订单服务不需要消费消息去下单
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                //其他
                return LocalTransactionState.UNKNOW;
            }
        });

        try {
            //4.启动producer
            transactionMQProducer.start();

            log.info("CreatePromoOrderProducer init end ...");
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        log.info("CreatePromoOrderProducer init failed ...");
    }


    public boolean sendTransactionMassage(CreateSeckillOrderRequest request, Long prductId, Long psId){

        //将request对象转化为json字符串
        String requestStr = JSON.toJSONString(request);

        //准备本地事务需要的参数
        HashMap<String, Long> parmMap = new HashMap<>();
        parmMap.put("productId", prductId);
        parmMap.put("psId", psId);

        //5.准备待发送的消息
        Message message = new Message();
        message.setTopic("create_promo_order");
        message.setBody(requestStr.getBytes(Charset.forName("utf-8")));

        //6.发送事务消息
        TransactionSendResult result = null;
        try {
            result = transactionMQProducer.sendMessageInTransaction(message, parmMap);
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        if (result != null && LocalTransactionState.COMMIT_MESSAGE.equals(result.getLocalTransactionState())) {
            return true;
        }
        return false;
    }
}
